package com.primeton.whdc.ccms.reportimpl.mqueue.consumer;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.primeton.whdc.ccms.reportimpl.entity.CallRecord;
import com.primeton.whdc.ccms.reportimpl.service.IApiCallRecordService;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author: YunTao.Li
 * @create: 2020-03-16 20:45
 * @description:
 **/
@Component
public class KafkaReceiver {
    //    @Autowired
//    private AsyncTask asyncTask;
    @Autowired
    private IApiCallRecordService apiCallRecordService;

    private Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);
    private Gson gson = new GsonBuilder().create();

//    @KafkaListener(topics = {"apicall"})
    public void receive(ConsumerRecord data) {
        System.out.println("app_log--消费消息:" + data.toString());

        try {
            // step1 : 还原实体对象
            CallRecord record = gson.fromJson(data.value().toString(), CallRecord.class);

//            String apiName = record.getApiName();
//            if (skip(apiName)) {
//
//            } else {
//                // step2 : 持久化
//
//            }
            apiCallRecordService.addAPICallRecord(record);
//            //TODO: step2 : 异步持久化
//            Future<String> result = asyncTask.insertRecord(record, apiCallRecordService);
//
//            if (result.isDone()) {
//                logger.info("插入数据成功 requestId:" + record.getRequestId());
//                result.get();
//            }
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
    }

    public static boolean skip(String key) {
        boolean f = false;
        if (StringUtils.isNotBlank(key) && key.indexOf("查询") != -1) {
            f = true;
        } else if (StringUtils.isNotBlank(key) && key.indexOf("获取下一个编号") != -1) {
            f = true;
        } else if (StringUtils.isNotBlank(key) && key.indexOf("list") == 0) {
            f = true;
        } else if (StringUtils.isNotBlank(key) && key.indexOf("获取") != -1) {
            f = true;
        } else if (StringUtils.isNotBlank(key) && key.indexOf("checkeds") == 0) {
            f = true;
        } else if (StringUtils.isNotBlank(key) && key.indexOf("query") != -1) {
            f = true;
        }

        return f;
    }


    public static void main(String[] args) {
        String key = "queryAll";
        boolean f = new KafkaReceiver().skip(key);

        System.out.println(f);
    }
}
